package com.showdor.springboot.jingqing;

import com.alibaba.fastjson2.util.DateUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.javafaker.Faker;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class JingQingProducer {

    private static final String TOPIC_NAME = "ycsj-jingqing-topic";
    // Kafka 服务器地址
    private static final String BOOTSTRAP_SERVERS = "192.168.139.185:32092";
    private static final Map<String, String> ROAD_MAP = new HashMap() {{
        put("1000", "京海高速0");
        put("1001", "京海高速1");
        put("1002", "京海高速2");
        put("1003", "京海高速3");
        put("1004", "京海高速4");
        put("1005", "京海高速5");
    }};
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final Producer<String, String> producer = new KafkaProducer<>(getKafkaProperties());
    static Faker faker = new Faker(new Locale("zh-CN"));

    public static void main(String[] args) throws InterruptedException {


        try {
            // 模拟过车数据
            int batch = 0;
            while (true) {
                sendVehicleData(batch);
                batch++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭 Kafka 生产者
            producer.close();
        }
    }

    private static Properties getKafkaProperties() {
        // 配置 Kafka 生产者
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    private static void sendVehicleData(int batch) throws InterruptedException, ExecutionException, JsonProcessingException {
        long startTime = System.currentTimeMillis();
        System.out.println("开始执行第" + batch + "次发送数据");

        int count = 0;
        while (true) {
            count++;
            Map<String, Object> vehicleData = generateJingQingData(faker, batch, count);
            String jsonData = objectMapper.writeValueAsString(vehicleData);
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "jg_key" + count, jsonData);

            producer.send(record);
            System.out.println("生成警情数据：" + count);
            Thread.sleep(50000); // 休眠 1 秒
        }
        // 确保每秒发送一条数据
    }


    private static Map<String, Object> generateJingQingData(Faker faker, int batch, int num) {

        Map<String, Object> jqMap = new HashMap<>();
        //按照上面的表字段及注释，填充jqMap,并根据表字段注释生成相应的模拟数据
        jqMap.put("num", num);
        jqMap.put("jjdbh", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("afdd", faker.address().streetAddress());
        jqMap.put("afxzqh", faker.address().city());
        jqMap.put("bjdh", faker.phoneNumber().phoneNumber());
        jqMap.put("bjfsdm", faker.regexify("[0-9]{1,2}"));
        jqMap.put("bjlbdm", faker.regexify("[0-9]{1,2}"));
        jqMap.put("bjlxdm", faker.regexify("[0-9]{1,2}"));
        jqMap.put("bjnr", faker.lorem().sentence());
        jqMap.put("bjrxb", faker.regexify("[男女]"));
        jqMap.put("bjrxm", faker.name().fullName());
        long now = System.currentTimeMillis();
        long bjsj = now - 1000 * 60 * faker.number().numberBetween(30, 60);
        jqMap.put("bjsj", DateUtils.format(bjsj));
        jqMap.put("cjlllbbh", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("dhdwjd", faker.address().longitude());
        jqMap.put("fxdwwd", faker.address().latitude());
        jqMap.put("gxdwdm", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("gxdwzt", faker.regexify("[0-9]{1,2}"));
        jqMap.put("hrsc", faker.number().numberBetween(1, 100));
        jqMap.put("hrsj", DateUtils.format(bjsj));
        //把呼终时间设置为呼入时间+随机时长
        jqMap.put("hzsj", DateUtils.format(bjsj + 1000 * 60 * faker.number().numberBetween(1, 10)));
        jqMap.put("jjlxdm", faker.regexify("[0-9]{1,2}"));
        jqMap.put("jjlyh", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("jjsc", faker.number().numberBetween(1, 100));
        jqMap.put("jjtbh", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("jjtip", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("jjybh", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("jjyxm", faker.name().fullName());
        jqMap.put("jqhm", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("jqlx", faker.regexify("[0-9]{1,2}"));
        jqMap.put("lxdh", faker.phoneNumber().phoneNumber());
        jqMap.put("sjdwdm", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("sjgxsj", DateUtils.format(now));
        jqMap.put("xzqh", faker.address().city());
        jqMap.put("yhdz", faker.address().streetAddress());
        jqMap.put("yhsfz", faker.regexify("[A-Z]{2}\\d{8}"));
        jqMap.put("yhxm", faker.name().fullName());
        jqMap.put("sfzdjq", faker.regexify("[0-9]{1,2}"));
        return jqMap;

    }

    /**
     * 生成车牌号
     */
    private static String generatePlateNumber(Faker faker) {
        // 生成一个随机的车牌号，例如 "浙A12345"
        return faker.regexify("[浙京津沪渝冀豫云辽黑湘皖鲁新苏浙赣鄂桂甘晋蒙陕吉闽贵粤青藏川宁琼][A-Z]\\d{5,6}");
    }

    /**
     * 车辆类型
     *
     * @return
     */
    private static String getVehicleClass(Integer num) {
        //从以逗号分隔的字符串中随机选择一个元素
        String[] elements = {
                "B11", "B12", "B13", "B14", "B15", "B16", "B17", "B18", "B19", "B1A", "B1B", "B1C", "B1D", "B1E", "B1F", "B1G", "B1H", "B1J", "B1K", "B1L", "B1U", "B1V", "B1W",
                "B21", "B22", "B23", "B24", "B25", "B26", "B27", "B28", "B29", "B2A", "B2B", "B2C", "B2D", "B2E", "B2F", "B2G", "B2H", "B2J", "B2K", "B2L", "B2U", "B2V", "B2W",
                "B31", "B32", "B33", "B34", "B35", "B36", "B37", "B38", "B39", "B3C", "B3D", "B3E", "B3F", "B3G", "B3H", "B3J", "B3K", "B3L", "B3U", "B3V", "B3W",
                "B41", "B42", "B43", "B44", "B45", "B46", "B47", "B48", "B49", "B4C", "B4D", "B4E", "B4F", "B4G", "B4H", "B4J", "B4K", "B4L", "B4U", "B4V", "B4W",
                "D11", "D12", "G11", "G12", "G13", "G14", "G15", "G16", "G17", "G18", "G19", "G1A", "G1B", "G1C", "G1D", "G1E", "G1F",
                "G21", "G22", "G23", "G24", "G25", "G26", "G27", "G28", "G29", "G2A", "G2B", "G2C", "G2D", "G2E", "G2F",
                "G31", "G32", "G33", "G34", "G35", "G36", "G37", "G38", "G3A", "G3B", "G3C", "G3D", "G3E", "G3F",
                "G41", "G42", "G43", "G44", "G45", "G46", "G48", "G4A", "G4B", "G4C", "G4D", "G4E", "G4F", "H11", "H12", "H13", "H14", "H15", "H16", "H17", "H18", "H19", "H1A", "H1B", "H1C", "H1D", "H1E", "H1F", "H1G", "H1J",
                "H21", "H22", "H23", "H24", "H25", "H26", "H27", "H28", "H29", "H2A", "H2B", "H2C", "H2D", "H2E", "H2F", "H2G", "H2J",
                "H31", "H32", "H33", "H34", "H35", "H37", "H38", "H39", "H3A", "H3B", "H3C", "H3D", "H3F", "H3G", "H3H", "H3J",
                "H41", "H42", "H43", "H44", "H45", "H46", "H47", "H4A", "H4B", "H4C", "H4F", "H4G", "H4H", "H4J",
                "H51", "H52", "H53", "H54", "H55", "H5B", "H5C",
                "J11", "J12", "J13",
                "K11", "K12", "K13", "K14", "K15", "K16", "K17", "K18", "K1A",
                "K21", "K22", "K23", "K24", "K25", "K26", "K27", "K28", "K2A",
                "K31", "K32", "K33", "K34", "K38", "K39", "K3A",
                "K41", "K42", "K43", "K49", "K4A",
                "M11", "M12", "M13", "M14", "M15"
        };
        String randomElement = elements[num % 6];
        return randomElement;
    }

    private static String getUsingPropertiesCode(Integer num) {
        return ("ABCDEFGHIJKLMNOPQRSTU").substring(num % 6, num % 6 + 1);
    }

    public static void createTopic(String bootstrapServers, String topicName, int numPartitions, short replicationFactor) {
        // 配置AdminClient
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // 创建AdminClient
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 定义新主题
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);

            // 创建主题
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();

            System.out.println("Topic '" + topicName + "' created successfully with " + numPartitions + " partitions.");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 检查主题是否存在，如果存在则删除
     */
    public static void deleteTopicIfExists(String bootstrapServers, String topicName) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 获取所有主题
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topics = topicsResult.names().get();

            // 检查主题是否存在
            if (topics.contains(topicName)) {
                System.out.println("Topic '" + topicName + "' exists. Deleting it...");

                // 删除主题
                DeleteTopicsResult deleteResult = adminClient.deleteTopics(Collections.singleton(topicName));
                deleteResult.all().get(); // 等待删除完成

                System.out.println("Topic '" + topicName + "' deleted successfully.");
            } else {
                System.out.println("Topic '" + topicName + "' does not exist.");
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
    /**
     * jjdbh	varchar	255					接警单编号
     * afdd	varchar	255					案发地点
     * afxzqh	varchar	255					案发行政区划
     * bjdh	varchar	255					报警电话
     * bjfsdm	varchar	255					报警方式代码
     * bjlbdm	varchar	255					报警类别代码
     * bjlxdm	varchar	255					报警类型代码
     * bjnr	varchar	255					报警内容
     * bjrxb	varchar	255					报警人性别
     * bjrxm	varchar	255					报警人姓名
     * bjsj	datetime					报警时间
     * cjlllbbh	varchar	255					处警力量列表编号
     * dhdwjd	double					电话定位经度
     * dhdwwd	double					电话定位纬度
     * fxdwjd	double					反向定位经度
     * fxdwwd	double					反向定位纬度
     * gxdwdm	varchar	255					管辖单位代码
     * gxdwzt	varchar	255					管辖单位状态
     * hrsc	int	11					呼入时长
     * hrsj	datetime					呼入时间
     * rksj	datetime					入库时间
     * hzsj	datetime					呼终时间
     * jjdcllxdm	varchar	255					接警单处理类型代码
     * jjdwdm	varchar	255					接警单位代码
     * jjdwmc	varchar	255					接警单位名称
     * jjdzt	varchar	255					接警单状态
     * jjlxdm	varchar	255					接警类型代码
     * jjlyh	varchar	255					接警录音号
     * jjsc	int	11					接警时长
     * jjtbh	varchar	255					接警台编号
     * jjtip	varchar	255					接警台ip
     * jjybh	varchar	255					接警员编号
     * jjyxm	varchar	255					接警员姓名
     * jqhm	varchar	255					警情号码
     * jqlx	varchar	255					警情类型
     * lxdh	varchar	255					联系电话
     * sjdwdm	varchar	255					受警单位代码
     * sjgxsj	datetime					数据更新时间
     * xzqh	varchar	255					行政区划
     * yhdz	varchar	255					用户地址
     * yhsfz	varchar	255					用户身份证
     * yhxm	varchar	255					用户姓名
     * sfzdjq	varchar	255					是否重大警情
     */
    /**
     * 按照上面的表字段及注释，生成java对象类，每个字段要生成get set方法及字段注释
     * **/
    public class JingQing {

    }
}